package com.saic.data.util;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.ObjectUtils.Null;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;

import oracle.net.aso.r;

import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

/**
 * @DESC:Hbase操作-->外部处理异常
 * @author Xinshiyou
 */
public class HbaseUtil {

	public static final String charSet = "UTF-8";
	public static final String hbaseNameSpace = "finance_platform";
	public static Configuration configuration;
	static {
		configuration = HBaseConfiguration.create();
		// JinQiao's Configuration
		configuration.set("hbase.zookeeper.property.clientPort", "2181");// master
		configuration.set("hbase.zookeeper.quorum","10.32.47.104,10.32.47.105,10.32.47.106");// Three
		configuration.set("hbase.master", "10.32.47.105:60000");// master
	}

	public static void createNameSpace(String spaceName)
			throws MasterNotRunningException, ZooKeeperConnectionException,
			IOException {
		HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);
		hBaseAdmin.createNamespace(NamespaceDescriptor.create("my_ns").build());
		hBaseAdmin.close();
	}
    
	@SuppressWarnings("resource")
	public static void addColumeFamily(String tableName,String familyName) {
		try {
			HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);
			tableName = addNameSpace(tableName);
			TableName tableNameObj = TableName.valueOf(tableName);
			HColumnDescriptor column=new HColumnDescriptor(familyName);
			hBaseAdmin.addColumn(tableNameObj, column);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * @创建数据表
	 * @param tableName
	 * @param list
	 * @throws IOException
	 * @throws ZooKeeperConnectionException
	 * @throws MasterNotRunningException
	 */
	public static void createTable(String tableName, List<String> list)
			throws MasterNotRunningException, ZooKeeperConnectionException,
			IOException {
		tableName = addNameSpace(tableName);
		HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);
		if (hBaseAdmin.tableExists(tableName)) {// If table exists, then
			hBaseAdmin.disableTable(tableName.getBytes(charSet));
			hBaseAdmin.deleteTable(tableName.getBytes(charSet));
		}
		System.out.println("HbaseUtil-createTable:" + tableName);
		HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
				for (String str : list)
			tableDescriptor.addFamily(new HColumnDescriptor(str
					.getBytes(charSet)));
		hBaseAdmin.createTable(tableDescriptor);
		if (null != hBaseAdmin)
			hBaseAdmin.close();
	}

	/**
	 * @DESC:添加HBASE空间
	 * @param tableName
	 */
	private static String addNameSpace(String tableName) {
		return hbaseNameSpace + ":" + tableName;
	}

	/**
	 * @重新创建HBASE表
	 * @param tableName
	 * @throws IOException
	 */
	public static void reCreateTable(String tableName, List<String> list)
			throws IOException {
		tableName = addNameSpace(tableName);
		HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);
		if (hBaseAdmin.tableExists(tableName)) {// If table exists, then
			hBaseAdmin.disableTable(tableName.getBytes(charSet));
			hBaseAdmin.deleteTable(tableName.getBytes(charSet));
		}
		System.out.println("HbaseUtil-createTable:" + tableName);
		HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
		for (String str : list)
			tableDescriptor.addFamily(new HColumnDescriptor(str
					.getBytes(charSet)));
		hBaseAdmin.createTable(tableDescriptor);
		if (null != hBaseAdmin)
			hBaseAdmin.close();
	}

	public static boolean updateData(String tableName, byte[] rowKey,
			String family, String column,String value) throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName.getBytes(charSet));
		Put put = new Put(rowKey);
		put.add(family.getBytes(charSet), column.getBytes(charSet), value.getBytes(charSet));
		table.put(put);
		table.flushCommits();
		table.close();
		return true;
	}

	/**
	 * @插入数据
	 * @param tableName
	 * @param rowKey
	 * @param map
	 * @throws IOException 
	 * @throws IOException
	 */
	public static Put getPutFromMap(String tableName,String family, String rowKey,
			Map<String, String> map) throws IOException {
		Put put = new Put(rowKey.getBytes(charSet));
		for (String key : map.keySet()) {
			put.add(family.getBytes(charSet), key.getBytes(charSet), map.get(key).getBytes(charSet));
		}		
		return put;
	}
	
	public static Put getPutFromResult(String tableName, Result r,
			String family,List<String> listColumnName) throws UnsupportedEncodingException {
			Put put = new Put(r.getRow());// TEMP TODO
			for (String name : listColumnName)
				put.add(family.getBytes(HbaseUtil.charSet),name.getBytes(HbaseUtil.charSet) ,
						r.getValue(family.getBytes(HbaseUtil.charSet),name.getBytes(HbaseUtil.charSet)));
			return put;	
	}

	public static boolean insertData(String tableName,String family, ResultScanner rs)
			throws IOException {
		List<Put> list = new ArrayList<Put>();
		for (Result r : rs) {
			Put put = new Put(r.getRow());
			 for(byte[] column:r.getFamilyMap(family.getBytes()).keySet())
					put.add(family.getBytes(), column, r.getFamilyMap(r.getRow()).get(column));
			list.add(put);
		}
        insertData(tableName, list);
		return true;
	}


	public static boolean insertData(String tableName, List<Put> list)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		table.put(list);
		table.flushCommits();
		table.close();
		return true;
	}
	
	public static boolean insertData(String tableName, Put put)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		table.put(put);
		table.flushCommits();
		table.close();
		return true;
	}

	/**
	 * @删除数据表
	 * @param tableName
	 * @throws IOException
	 * @throws ZooKeeperConnectionException
	 * @throws MasterNotRunningException
	 */
	public static void dropTable(String tableName)
			throws MasterNotRunningException, ZooKeeperConnectionException,
			IOException {
		tableName = addNameSpace(tableName);
		HBaseAdmin admin = new HBaseAdmin(configuration);
		admin.disableTable(tableName);
		admin.deleteTable(tableName);
		admin.close();
	}

	public static void deleteRowAll(String tableName)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		ResultScanner rScanner = table.getScanner(new Scan());
		List<Delete> dList = new ArrayList<Delete>();
		for(Result result:rScanner){
			Delete d = new Delete(result.getRow());
			dList.add(d);
		}
		table.delete(dList);
		table.close();
	}
	/**
	 * @删除一行数据
	 * @param tablename
	 * @param rowkey
	 * @throws IOException
	 */
	public static boolean deleteRow(String tableName, String rowkey)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		Delete d = new Delete(rowkey.getBytes());
		table.delete(d);
		table.close();
		return true;
	}
	/**
	 * @根据Row删除行
	 * @param tablename
	 * @param rowkey
	 * @return
	 * @throws IOException
	 */
	public static boolean deleteRow(String tableName, byte[] rowkey)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		Delete d = new Delete(rowkey);
		table.delete(d);
		table.close();
		return true;
	}
	/**
	 * @全表删除列
	 * @param tablename
	 * @param rowkey
	 * @return
	 * @throws IOException
	 */
	public static boolean deleteColumnAll(String tableName,String family,String column)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		ResultScanner rScanner =table.getScanner(new Scan());
		for(Result r: rScanner){
			Delete d = new Delete(r.getRow()).deleteColumns(family.getBytes(), column.getBytes());
			table.delete(d);
		}
		table.close();
		return true;
	}
	
	
    
	/**
	 * @根据Row删除列
	 * @param tablename
	 * @param rowkey
	 * @return
	 * @throws IOException
	 */
	public static boolean deleteColumn(String tableName, byte[] rowkey,String family,String column)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		Delete d = new Delete(rowkey).deleteColumns(family.getBytes(), column.getBytes());
		table.delete(d);
		table.close();
		return true;
	}
	
	/**
	 * @列拷贝
	 * @param tablename
	 * @param rowkey
	 * @return
	 * @throws IOException
	 */
	public static boolean columnCopy(String tableName,String family,String column1,String column2)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		ResultScanner rScanner =table.getScanner(new Scan());
		for(Result r: rScanner){
			Put put = new Put(r.getRow());
			put.add(family.getBytes(), column2.getBytes(),r.getValue(family.getBytes(), column1.getBytes()));
			table.put(put);
		}
		table.flushCommits();
		table.close();
		return true;
	}
	/**
	 * @列更新
	 * @param tablename
	 * @param rowkey
	 * @return
	 * @throws IOException
	 */
	public static boolean updateColumn(String tableName,String familyName,String colName,String valueBefore,String valueAfter)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = null;
		table = new HTable(configuration, tableName);
		Filter filter = new SingleColumnValueFilter(familyName.getBytes(charSet),
				colName.getBytes(charSet), CompareOp.EQUAL, valueBefore.getBytes(charSet));
		Scan s = new Scan();
		s.setFilter(filter);
		ResultScanner rScanner= table.getScanner(s);
		List<Put> puts = new ArrayList<Put>(); 
		for(Result r: rScanner){
			byte[] rowKey = r.getRow();
			Put put = new Put(rowKey);
			put.add(familyName.getBytes(charSet), colName.getBytes(charSet), valueAfter.getBytes(charSet));
		    puts.add(put);
		}
		table.put(puts);
		table.flushCommits();
		table.close();
		return true;
		
	}
	
	/**
	 * @全表扫描删除列种换行符
	 * @param tablename
	 * @param family
	 * @param column
	 * @return
	 * @throws IOException
	 */
	public static boolean columnReplace(String tableName,String family,String column,String valueBefore,String valueAfter)
			throws IOException {
		//tableName = addNameSpace(tableName);
		if (valueBefore == null || valueAfter == null) return false;
		HTable table = new HTable(configuration, tableName);	
		ResultScanner rScanner =table.getScanner(new Scan());
		List<Put> puts =  new ArrayList<Put>();
		for(Result r: rScanner){
			String value = getFromResult(r, family, column);
			if (value==null) continue;
			if (value.contains("\r\n")) {
			   Put put = new Put(r.getRow());
			   put.add(family.getBytes(), column.getBytes(), value.replace(valueBefore, valueAfter).getBytes());
			   puts.add(put);
			}
		} 
		if (puts.size()>0) table.put(puts);
		table.close();
		return true;
	}
	
	/**
	 * @根据Row查询行：需要关闭Result
	 * @param tableName
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	@SuppressWarnings({ "resource" })
	public static Result QueryByConditionKey(String tableName, String rowKey)
			throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = new HTable(configuration, tableName);
		Get scan = new Get(rowKey.getBytes());
		return table.get(scan);
	}

	/**
	 * @根据Row、Column Name查询数据
	 * @param tableName
	 * @param rowKey
	 * @param familyName
	 * @param columnName
	 * @return
	 * @throws IOException
	 */
	public static String QueryByKeyName(String tableName, byte[] rowKey,
			String familyName,String columnName) throws IOException {
		Result res = QueryByConditionKey(tableName, new String(rowKey, charSet));
		return new String(res.getValue(familyName.getBytes(charSet), columnName.getBytes(charSet)),charSet);
	}
	
	/**
	 * @根据列名扫描全表
	 * @param tableName
	 * @param familyName
	 * @param colName
	 * @param colValue
	 * @return
	 * @throws IOException
	 */
	@SuppressWarnings({ "resource" })
	public static ResultScanner QueryByConditionValue(String tableName,
			String familyName,String colName, String colValue) throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = null;
		table = new HTable(configuration, tableName);
		Filter filter = new SingleColumnValueFilter(familyName.getBytes(charSet),
				colName.getBytes(charSet), CompareOp.EQUAL, colValue.getBytes(charSet));
		Scan s = new Scan();
		s.setFilter(filter);
		return table.getScanner(s);
	}
	
	/**
	 * @根据列名扫描全表
	 * @param tableName
	 * @param familyName
	 * @param colName
	 * @param colValue
	 * @return
	 * @throws IOException
	 */
	@SuppressWarnings({ "resource" })
	public static ResultScanner QueryByConditionValue(String tableName,
			String familyName,String colName, String colValueFrom, String colValueTo) throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = null;
		table = new HTable(configuration, tableName);
		Filter filterF = new SingleColumnValueFilter(familyName.getBytes(charSet),
				colName.getBytes(charSet), CompareOp.GREATER_OR_EQUAL, colValueFrom.getBytes(charSet));
		Filter filterT = new SingleColumnValueFilter(familyName.getBytes(charSet),
				colName.getBytes(charSet), CompareOp.GREATER_OR_EQUAL, colValueFrom.getBytes(charSet));
		FilterList fList = new FilterList();
		fList.addFilter(filterF);
		fList.addFilter(filterT);
		Scan s = new Scan();
		s.setFilter(fList);
		return table.getScanner(s);
	}

	/**
	 * @根据索引范围查找数据
	 * @param tableName
	 * @param from
	 * @param to
	 * @return
	 * @throws IOException
	 */
	@SuppressWarnings({ "resource" })
	public static ResultScanner QueryByCondition(String tableName,
			String startRow, String stopRow) throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = null;
		table = new HTable(configuration, tableName);
		Scan s = new Scan();
		s.setStartRow(startRow.getBytes(charSet));
		s.setStopRow(stopRow.getBytes(charSet));
		return table.getScanner(s);
	}

	/**
	 * @根据表名扫描全表
	 * @param tableName
	 * @return
	 * @throws IOException
	 */
	@SuppressWarnings({ "resource" })
	public static ResultScanner scanHbase(String tableName) throws IOException {
		tableName = addNameSpace(tableName);
		HTable table = null;
		table = new HTable(configuration, tableName);
		return table.getScanner(new Scan());
	}
	
	/**
	 * @获取列值
	 * @param rs 
	 * @param colName
	 * @return
	 * @throws IOException
	 */
	public static String getFromResult(Result rs, String family,String colName)
			throws UnsupportedEncodingException {
		String rstr =null;
		byte[] rb =rs.getValue(family.getBytes(charSet), colName.getBytes(charSet));
		if (null!=rb) rstr = new String(rb,charSet);
		else rstr="";
		return rstr;
	}
	
	/**
	 * @规范列值
	 * @param input 
	 * @return
	 * @throws IOException
	 */
	public static String format(String input) {
		if (input.endsWith("-"))// Check '-' position
			input = "-" + input.substring(0, input.length() - 1);
		String output = input;
		if (input.contains(","))// Delete ',' from input values
			output = output.replace(",", "");
		// Other checks
		return output.trim();
	}
    
	@SuppressWarnings("resource")
	public static List<String> getFamilyColumns(String tableName,String family) throws Exception{
		 List<String> cList = new ArrayList<String>();
		 HTable table = null;
		 table = new HTable(configuration, tableName);
		 ResultScanner rs =table.getScanner(new Scan());
		 Result result = rs.next();
		 for(byte[] column :result.getFamilyMap(family.getBytes()).keySet())
			 cList.add(family+":"+new String(column));
		 return cList;
	}

}